
#include <iostream>

#include <boost/filesystem.hpp>
#include <boost/algorithm/string.hpp>

#include <boost/bind.hpp>
#include <boost/thread.hpp>
#include <boost/lexical_cast.hpp>
#include <boost/filesystem.hpp>

#include <boost/program_options.hpp>
namespace po = boost::program_options;

#include "bocat/common/logging.hpp"
#include "bocat/common/time.hpp"

#include "boque/trasit/channel.hpp"


static  bocat::common::Logger*  logger_p = NULL;

#define INFO(ARGS...)   logger_p->info(__LINE__, __FILE__, ARGS)
#define ERROR(ARGS...)  logger_p->error(__LINE__, __FILE__, ARGS)


namespace boque		{
namespace trasit	{


///////////////////////////////////////////////////////////////////////
//

class	Watcher
{
	typedef	Channel*	channel_ptr;
	typedef	std::map< std::string, channel_ptr >	channels_type;
	typedef	channels_type::iterator		channels_iterator_type;

public	:
	Watcher( const char* data_root ) 
		:	_data_root( data_root ), 
			_will_stop( false ),
			_running( true )
	{
		_channels_p = &_channels;
	}

	~Watcher()
	{
	}

public	:
	void	run( void )
	{
		_running = true;

		while ( !_will_stop ) {
			// scan data-root, to find new channel
			scanDataRoot();

			// scan channels, to create new waiter
			for ( channels_iterator_type it = _channels_p->begin();
				  _channels_p->end() != it;
				  ++it ) {
				const std::string&	channel_name = it->first;
				Channel*	channel_p = (Channel *)it->second;

				if ( channel_p->inAliveMode() ) {
					//INFO( " - check channel %s in ALIVE mode ...", channel_name.c_str() );
					if ( channel_p->emptyQueue() ) {
						if ( NULL != channel_p->getWaiter() ) {
							INFO( " - clear channel %s waiter", channel_name.c_str() );
							if ( channel_p->isRelaying() ) {
								INFO( "         channel %s is still relaying ...", channel_name.c_str() );
							}

							channel_p->disableRelay();

							boost::thread*	tp = channel_p->getWaiter();
							tp->join();
							delete tp;

							channel_p->setWaiter( NULL );
						}
					}
					else {
						INFO( " - channel %s enter BUFFER mode ...", channel_name.c_str() );
						channel_p->inBufferedMode( true );
					}
				}

				if ( channel_p->inBufferedMode() ) {
					if ( !channel_p->isRelaying() ) {
						if ( NULL == channel_p->getWaiter() ) {
							boost::thread*	tp = new boost::thread( boost::bind(&Channel::relay, channel_p) );

							channel_p->setWaiter( tp );
						}

						channel_p->enableRelay();
					}
				}
				else {
					// may be in UNKNOWN mode
					if ( !channel_p->emptyQueue() ) {
						INFO( " - channel %s enter BUFFER mode ...", channel_name.c_str() );
						channel_p->inBufferedMode( true );
					}
				}
			}

			boost::this_thread::sleep(boost::posix_time::milliseconds(1000));
			//INFO( "  - watcher routine (will_stop = %d) ...", _will_stop );
		}	//// while ( !_will_stop )

		// stop all relaying
		for ( channels_iterator_type it = _channels_p->begin();
			  _channels_p->end() != it;
			  ++it ) {
			const std::string&	channel_name = it->first;
			Channel*	channel_p = (Channel *)it->second;

			if ( channel_p->isRelaying() ) {
				INFO( "         channel %s is still relaying ...", channel_name.c_str() );
			}
			while ( channel_p->isRelaying() ) {
				channel_p->disableRelay();
				boost::this_thread::sleep(boost::posix_time::milliseconds(100));
			}
		}

		_running = false;
	}

	void	stop( void )
	{
		INFO( " * watcher get stop signal, and will stop ..." );
		_will_stop = true;
	}

	bool	running( void )
	{
		return	_running;
	}
	
private	:
	void	scanDataRoot( void )
	{
		boost::filesystem::path data_path( _data_root );

		if ( !boost::filesystem::exists( data_path ) ) {
			try {
				boost::filesystem::create_directories( data_path );
			}
			catch( std::exception& e ) {
				return;
			}
		}

		boost::filesystem::directory_iterator end_itr;

		for ( boost::filesystem::directory_iterator itr( data_path );
			itr != end_itr;
			++itr ) { 
			if (boost::filesystem::is_directory( itr->status() ) ) { 
				std::string	dirname = itr->path().filename();
				std::string	to_s = dirname;
				restoreAddressFormat( to_s );
				boost::trim( to_s );

				if ( _channels_p->count( to_s ) < 1 ) {
					std::string	channel_dir = itr->path().string();

					Channel*    channel_p = new Channel( to_s, channel_dir );
					_channels_p->insert( std::make_pair( to_s, channel_p ) );

					INFO( " channel %s <-- %s found", to_s.c_str(), channel_dir.c_str() );
				}

			}
		}

	}


	void	restoreAddressFormat( std::string& address )
	{
		boost::replace_all( address, "%%%", "://" );
		boost::replace_all( address, "%", ":" );
	}

private	:
	const char*	_data_root;

	channels_type	_channels;
	channels_type*	_channels_p;

	bool	_will_stop;
	bool	_running;
};


}	//// namespace trasit
}	//// namespace boque



int main(int argc, char* argv[])
{
    std::string data_root;

    std::string log_config;
    std::string log_section;

    try {
        po::options_description desc("Allowed options");
        desc.add_options()
            ("help,h", "produce help message")
            ("data-root", po::value<std::string>(&data_root), "data root to watch")
            //("threads", po::value<int>(&thread_count)->default_value(1), "number of threads to run")

            ("log-config", po::value<std::string>(&log_config), "path to logging config file")
            ("log-section", po::value<std::string>(&log_section), "logging section name in config file")
        ;

        po::variables_map vm;
        po::store(po::parse_command_line(argc, argv, desc), vm);
        po::notify(vm);

        if (vm.count("help")) {
            std::cout << desc << "\n";
            return 1;
        }

        if ((vm.count("log-config") > 0) && (vm.count("log-section") < 1)) {
            std::cout << "\nlog-section must be provided\n\n";
            std::cout << desc << "\n";
            return 2;
        }
    }
    catch(std::exception& e) {
        std::cerr << "error: " << e.what() << "\n";
        return 1;
    }
    catch(...) {
        std::cerr << "Exception of unknown type!\n";
        return 2;
    }


    logger_p = bocat::common::getLogger();
    if ( log_config.size() > 0 ) {
		boost::filesystem::path log_conf_path( log_config );
		if ( !boost::filesystem::exists( log_conf_path ) ) {
			std::cerr << "config file " << log_config << " does NOT exist.\n";
			return 3;
		}
		if ( !boost::filesystem::is_regular_file( log_conf_path ) ) {
			std::cerr << "config file " << log_config << " is NOT a regular file.\n";
			return 4;
		}

        logger_p->configure( log_config.c_str(), log_section.c_str() );
    }

    try
    {
        // Block all signals for background thread.
        sigset_t new_mask;
        sigfillset(&new_mask);
        sigset_t old_mask;
        pthread_sigmask(SIG_BLOCK, &new_mask, &old_mask);

        INFO(" * initializing ...");

		boque::trasit::Watcher*	watcher_p = new boque::trasit::Watcher( data_root.c_str() );

		boost::thread*	watcher_tp = new boost::thread( boost::bind(&boque::trasit::Watcher::run, watcher_p) );

        // Restore previous signals.
        pthread_sigmask(SIG_SETMASK, &old_mask, 0);

        // Wait for signal indicating time to shut down.
        sigset_t wait_mask;
        sigemptyset(&wait_mask);
        sigaddset(&wait_mask, SIGINT);
        sigaddset(&wait_mask, SIGQUIT);
        sigaddset(&wait_mask, SIGTERM);
        pthread_sigmask(SIG_BLOCK, &wait_mask, 0);
        int sig = 0;
        sigwait(&wait_mask, &sig);

        // Stop the server.
		while ( watcher_p->running() ) {
			boost::this_thread::sleep(boost::posix_time::milliseconds(100));
			watcher_p->stop();
		}
		delete watcher_p;

		watcher_tp->join();
		delete watcher_tp;

        INFO("    shutdown ...");
    }
    catch (std::exception& e)
    {
        ERROR("exception: %s", e.what());
		return	1;
    }

    INFO("    done.");

    return 0;
}



